package day04

import org.apache.spark.rdd.RDD
import org.apache.spark.{HashPartitioner, Partitioner, SparkConf, SparkContext}

/**
 * 2.4.3	Key-Value类型
 */
object TransformationKeyValue {

}

/**
 * 2.4.3.1 partitionBy()按照K重新分区
 */
object Transformation_KeyValue_partitionBy {

  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val rdd: RDD[(Int, String)] = sc.makeRDD(Array((1, "aaa"), (2, "bbb"), (3, "ccc")), 3)

    // 打印重新分区前的数据
    rdd.mapPartitionsWithIndex((index, data) => {
      println(index + "号分区：" + data.mkString(", "))
      data
    }).collect()

    // 调用partitionBy对RDD重新分区
    val newRdd: RDD[(Int, String)] = rdd.partitionBy(new HashPartitioner(2))

    // 打印重新分区后的数据
    newRdd.mapPartitionsWithIndex((index, datas) => {
      println(index + "号分区：" + datas.mkString(", "))
      datas
    }).collect()

    // 关闭连接
    sc.stop()
  }
}

/**
 * 自定义分区器
 */
object Transformation_KeyValue_custom_partitionBy {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val rdd: RDD[(Int, String)] = sc.makeRDD(Array((1, "aaa"), (2, "bbb"), (3, "ccc")), 3)

    // 打印重新分区前的数据
    rdd.mapPartitionsWithIndex((index, datas) => {
      println(index + "号分区：" + datas.mkString(", "))
      datas
    }).collect()

    // 调用partitionBy进行重新分区，传入自定义的分区
    val newRdd: RDD[(Int, String)] = rdd.partitionBy(new MyPartitioner(2))

    newRdd.mapPartitionsWithIndex((index, datas) => {
      println(index + "号分区：" + datas.mkString(", "))
      datas
    }).collect()

    // 关闭连接
    sc.stop()
  }
}

class MyPartitioner(num: Int) extends Partitioner {
  override def numPartitions: Int = num

  override def getPartition(key: Any): Int = {
    key match {
      case i: Int =>
        if (i % 2 == 0) {
          0
        } else {
          1
        }
      case _ =>
        0
    }
  }

  override def equals(other: Any): Boolean = other match {
    case h: MyPartitioner =>
      h.numPartitions == numPartitions
    case _ =>
      false
  }

  override def hashCode: Int = numPartitions
}

/**
 * 2.4.3.2 reduceByKey()按照K聚合V
 */
object Transformation_KeyValue_reduceByKey {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("b", 5), ("c", 2), ("b", 3), ("a", 4)))

    // 调用reduceByKey()
    val newRdd: RDD[(String, Int)] = rdd.reduceByKey((x, y) => x + y)

    // 打印结果
    newRdd.collect().foreach(println)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.4.3.3 groupByKey()按照K重新分组
 */
object Transformation_KeyValue_groupByKey {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("b", 5), ("a", 5), ("b", 2)))

    // 调用groupByKey()，将相同key对应值聚合到一个Seq中
    val newRdd: RDD[(String, Iterable[Int])] = rdd.groupByKey()

    // 打印输出
    newRdd.collect().foreach(println)

    // 计算相同key对应值的相加结果
    val newRdd1: RDD[(String, Int)] = newRdd.map(t => (t._1, t._2.sum))

    newRdd1.collect().foreach(println)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.4.3.4 reduceByKey和groupByKey区别
 *
 * 1. reduceByKey：按照Key进行聚合，在shuffle之前有combine(预聚合)操作，返回结果是RDD[k,v]
 *
 * 2. groupByKey：按照key进行分组，直接进行shuffle
 *
 * 3. 开发指导：在不影响业务逻辑的前提下，优先选用reduceByKey。求和操作不影响业务逻辑，求平均值影响业务逻辑
 *
 */
object Transformation_KeyValue_reduceByKey_diff_groupByKey {

}

/**
 * 2.4.3.5 aggregateByKey()按照K处理分区内和分区间逻辑
 */
object Transformation_KeyValue_aggregateByKey {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 3), ("a", 2), ("c", 4), ("b", 3), ("c", 6), ("c", 8)), 2)

    // 调用aggregateByKey()，初始值0，分区内取最大值，分区间相加
    val resRDD: RDD[(String, Int)] = rdd.aggregateByKey(0)(Math.max, _ + _)

    // 打印输出
    resRDD.collect().foreach(println)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.4.3.6 foldByKey()分区内和分区间相同的aggregateByKey()
 */
object Transformation_KeyValue_foldByKey {

  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val list: List[(String, Int)] = List(("a", 1), ("a", 1), ("a", 1), ("b", 1), ("b", 1), ("b", 1), ("b", 1), ("a", 1))
    val rdd: RDD[(String, Int)] = sc.makeRDD(list, 2)

    // 调用aggregateByKey()，分区内和分区间的计算逻辑相同
    val newRdd: RDD[(String, Int)] = rdd.aggregateByKey(0)(_ + _, _ + _)

    // 当分区内和分区间的计算规则相同时，可以使用foldByKey
    val resRDD: RDD[(String, Int)] = rdd.foldByKey(0)(_ + _)

    newRdd.collect().foreach(println)

    println("------------------------------")

    resRDD.collect().foreach(println)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.4.3.7 combineByKey()转换结构后分区内和分区间操作
 */
object Transformation_KeyValue_combineByKey {

  // 需求说明：创建一个pairRDD，根据key计算每种key的均值。（先计算每个key对应值的总和以及key出现的次数，再相除得到结果）

  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    val list: List[(String, Int)] = List(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98))

    val rdd: RDD[(String, Int)] = sc.makeRDD(list, 2)

    // 调用combineByKey()，将数据个数转换成tuple类型，分区内V相加，count+1；分区间值相加
    val resRdd: RDD[(String, (Int, Int))] = rdd.combineByKey((_, 1),
      (acc: (Int, Int), v: Int) => (acc._1 + v, acc._2 + 1),
      (acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2))

    // 打印输出
    resRdd.collect().foreach(println)

    // 求平均值
    val avgRdd: RDD[(String, Double)] = resRdd.map {
      case (key, value) => (key, value._1 / value._2.toDouble)
    }

    // 打印输出
    avgRdd.collect().foreach(println)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.4.3.8 reduceByKey、aggregateByKey、foldByKey、combineByKey 的区别
 */
object Transformation_KeyValue_diff {

  def main(args: Array[String]): Unit = {

    // reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)]
    // 等价于 combineByKeyWithClassTag[V]((v: V) => v, func, func, partitioner)  ---> 第一个初始值不变，分区内和分区间处理规则一致

    // aggregateByKey[U: ClassTag](zeroValue: U, partitioner: Partitioner)(seqOp: (U, V) => U,combOp: (U, U) => U): RDD[(K, U)]
    // 等价于 combineByKeyWithClassTag[U]((v: V) => cleanedSeqOp(createZero(), v),cleanedSeqOp, combOp, partitioner) ---> 指定初始值，第一个初始值和分区内处理规则一致

    //foldByKey(zeroValue: V,partitioner: Partitioner)(func: (V, V) => V): RDD[(K, V)]
    // 等价于 combineByKeyWithClassTag[V]((v: V) => cleanedFunc(createZero(), v),cleanedFunc, cleanedFunc, partitioner) ---> 指定初始值，第一个初始值、分区内、分区间的规则一致

    // combineByKey[C](createCombiner: V => C,mergeValue: (C, V) => C,mergeCombiners: (C, C) => C,partitioner: Partitioner,mapSideCombine: Boolean = true,serializer: Serializer = null): RDD[(K, C)]
    // 等价于 combineByKeyWithClassTag(createCombiner, mergeValue, mergeCombiners,partitioner, mapSideCombine, serializer)(null) ---> 把第一个值变成特定的结构
  }
}

/**
 * 2.4.3.9 sortByKey()按照K进行排序
 */
object Transformation_KeyValue_sortByKey {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val rdd: RDD[(Int, String)] = sc.makeRDD(Array((3, "aa"), (6, "cc"), (2, "bb"), (1, "dd")), 3)

    println("------------------------排序之前---------------------------")

    rdd.mapPartitionsWithIndex((index, data) => {
      var str: String = "";
      for (elem <- data) {
        str += elem._1 + "->" + elem._2 + ";"
      }
      println(index + "号分区：" + str)
      data
    }).collect();

    println("------------------------排序以及重新分区后---------------------------")

    // 调用sortByKey进行降序排序，并重新分成2个区
    val sortedRdd: RDD[(Int, String)] = rdd.sortByKey(ascending = false, 2)

    sortedRdd.mapPartitionsWithIndex((index, data) => {
      var str: String = "";
      for (elem <- data) {
        str += elem._1 + "->" + elem._2 + ";"
      }
      println(index + "号分区：" + str)
      data
    }).collect()

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.4.3.9 mapValues()只对V进行操作
 */
object Transformation_KeyValue_mapValues {

  def main(args: Array[String]): Unit = {

    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val rdd: RDD[(Int, String)] = sc.makeRDD(Array((1, "a"), (1, "d"), (2, "b"), (3, "c")))

    // 对V进行操作
    rdd.mapValues(_ + "|||").collect().foreach(println)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.4.3.10 join()连接 将相同key对应的多个value关联在一起
 *
 * 如果key只是某一个RDD有，这个key不会关联
 */
object Transformation_KeyValue_join {

  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.1 创建第一个RDD
    val rdd1: RDD[(Int, String)] = sc.makeRDD(Array((1, "a"), (2, "b"), (3, "c")))

    //3.2 创建第二个pairRDD
    val rdd2: RDD[(Int, Int)] = sc.makeRDD(Array((1, 4), (2, 5), (4, 6)))

    val rdd1JoinRdd2: RDD[(Int, (String, Int))] = rdd1.join(rdd2)
    rdd1JoinRdd2.collect().foreach(println)

    println("-------------------------------------------------")

    val rdd2JoinRdd1: RDD[(Int, (Int, String))] = rdd2.join(rdd1)
    rdd2JoinRdd1.collect().foreach(println)


    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.4.3.11 cogroup() 类似全连接，但是在同一个RDD中对key聚合
 *
 * 操作两个RDD中的KV元素，每个RDD中相同key中的元素分别聚合成一个集合。
 */
object Transformation_KeyValue_cogroup {

  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val rdd1: RDD[(Int, String)] = sc.makeRDD(Array((1, "a"), (2, "b"), (3, "c")))
    val rdd2: RDD[(Int, Int)] = sc.makeRDD(Array((1, 4), (2, 5), (3, 6)))

    // 调用cogroup()
    val coGroupRdd: RDD[(Int, (Iterable[String], Iterable[String], Iterable[Int]))] = rdd1.cogroup(rdd1, rdd2)

    coGroupRdd.collect().foreach(println)

    // 关闭连接
    sc.stop()
  }
}







